package com.bw.dws;

import net.bwie.realtime.jtp.DorisUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsEntranceBehavior {
    public static void main(String[] args) throws Exception {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);


        // 读取数据 设置水位线
        tEnv.executeSql("CREATE TABLE dwd_entrance_guard_log (\n" +
                "id STRING,\n" +
                "ownerId INT,\n" +
                "openTime STRING,\n" +
                "cId INT,\n" +
                "houseCode STRING,\n" +
                "roomCode STRING,\n" +
                "openMode SMALLINT,\n" +
                "districtName STRING,\n" +
                // 水位线
                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(openTime),0),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '1' SECOND" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_entrance_guard_log',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'dwd_entrance_guard_log_group',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")"
        );

        // 窗口计算
        TableResult window = tEnv.executeSql(
                "select  " +
                        "TUMBLE_START(time_ltz,INTERVAL '1' second) as stt," +
                        "TUMBLE_END(time_ltz,INTERVAL '1' second) as edt," +

                        // 小区ID
                        "cId," +
                        // 业主id
                        "ownerId," +
                        // 楼栋编号
                        "houseCode," +
                        // 房间编号
                        "roomCode," +
                        // 地区名称
                        "districtName,"+

                        "  SUM(CASE WHEN openMode = 1 THEN 1 ELSE 0 END) AS card_cnt,\n" +
                        "  SUM(CASE WHEN openMode = 2 THEN 1 ELSE 0 END) AS pw_cnt,\n" +
                        "  SUM(CASE WHEN openMode = 3 THEN 1 ELSE 0 END) AS face_cnt,\n" +
                        "  SUM(CASE WHEN openMode = 4 THEN 1 ELSE 0 END) AS remote_cnt\n" +
                        " from dwd_entrance_guard_log " +
                        "WHERE cId IS NOT  null \n" +
                        "GROUP BY cId,ownerId,houseCode,roomCode, TUMBLE(time_ltz,INTERVAL '1' second) "
        );
        window.print();

        tEnv.createTemporaryView( "windowTable", (Table) window);

        tEnv.executeSql(
                "CREATE TABLE dws_owner_behavior (\n" +

                "stt VARCHAR(225) COMMENT '窗口开始日期时间' ,\n" +
                "edt VARCHAR(225) COMMENT '窗口结束日期时间' ,\n" +

                "cId INT COMMENT '小区ID' ,\n" +
                "ownerId INT COMMENT '业主id' ,\n" +
                "houseCode VARCHAR(225) COMMENT '楼栋编号' ,\n" +
                "roomCode VARCHAR(225) COMMENT '房间编号' ,\n" +
                "districtName VARCHAR(225) COMMENT '地区名称' ,\n" +

                "card_cnt BIGINT COMMENT '刷卡次数' ,\n" +
                "pw_cnt BIGINT COMMENT '密码次数' ,\n" +
                "face_cnt BIGINT COMMENT '人脸次数' ,\n" +
                "remote_cnt BIGINT COMMENT '远程次数'  \n" +

                ")WITH (\n" +
                " 'connector' = 'doris',\n" +
                " 'fenodes' = 'node102:8030',\n" +
                " 'table.identifier' = 'sca_doris.dws_owner_behavior',\n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'sink.label-prefix' = 'doris_label'\n" +
                ");"
        );

        tEnv.executeSql("insert into dws_owner_behavior \n" +
                "select \n" +
                "CAST(stt AS VARCHAR(225)) as stt , \n" +
                "CAST(edt AS VARCHAR(225)) as edt , \n" +

                 "cId, \n" +
                 "ownerId, \n" +
                 "houseCode, \n" +
                 "roomCode, \n" +
                 "districtName, \n" +

                 "card_cnt, \n" +
                 "pw_cnt, \n" +
                 "face_cnt, \n" +
                 "remote_cnt \n" +
                " from windowTable");


        // 提交Flink作业
//        senv.execute("DwsOwnerBehavior");
    }


}
